package com.goxplanet;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

public class StockRiskJob {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStream<StockMarketPrice> priceStream = env
                .addSource(new StockMarketSource())
                .name("prices")
                ;
        DataStream<StockTrade> tradesStream = env
                .addSource(new StockTradeSource())
                .name("trades")
//                .assignTimestampsAndWatermarks(
//                        WatermarkStrategy.<StockTrade>forBoundedOutOfOrderness(Duration.ofSeconds(5))
//                                .withTimestampAssigner((event, timestamp) -> event.timestamp)
//                )
                ;

        DataStream<StockPosition> positionStream = tradesStream
                .keyBy(r -> r.ticker)
				.process(new StockTradeProcessFunction());

        DataStream<StockRisk> riskStream = priceStream.keyBy(r -> r.ticker).connect(positionStream.keyBy(r -> r.ticker))
                .process(new StockRiskProcessFunction());

        tradesStream.print();
        positionStream.print();
        riskStream.print();

        env.execute("Flink Java API Skeleton");
    }
}
